package team.bluepen.supermarket.service;

import org.springframework.stereotype.Service;
import team.bluepen.supermarket.conf.KafkaProperties;
import team.bluepen.supermarket.data.entity.Product;
import team.bluepen.supermarket.kafka.KafkaReceiveCallback;
import team.bluepen.supermarket.kafka.KafkaReceiver;
import team.bluepen.supermarket.kafka.KafkaTopics;

import java.util.ArrayList;
import java.util.Deque;
import java.util.List;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * @author RollW
 */
@Service
public class ProductRecvService implements KafkaReceiveCallback<Product> {
    private final Deque<Product> productDeque = new LinkedBlockingDeque<>(500);
    private final KafkaReceiver<Product> receiver;

    public ProductRecvService(KafkaProperties kafkaProperties) {
        receiver = new KafkaReceiver<>(KafkaTopics.PRODUCT_TOPIC,
                kafkaProperties.getBootstrapServers(),
                Product.deserializer(), this);
        receiver.start();
    }

    public List<Product> getReceived() {
        return new ArrayList<>(productDeque);
    }

    @Override
    public void onReceive(Product data) {
        while (productDeque.size() >= 500) {
            productDeque.remove();
        }
        productDeque.push(data);
    }
}
